package thread;

import java.util.concurrent.Executor;


public class SendRunnable  extends AbstractQueueRunnable<Object> {


	/**
	 *
	 * @param channelContext
	 * @param executor
	 * @author tanyaowu
	 */
	public SendRunnable( Executor executor) {
		super(executor);
	}

	/**
	 *
	 */
	@Override
	public boolean addMsg(Object obj) {
		if (this.isCanceled()) {
			System.out.println("{}, 任务已经取消，{}添加到发送队列失败");
			return false;
		}
		System.out.println("放入队列 = "+obj);
		return msgQueue.add(obj);
	}

	/**
	 * 清空消息队列
	 */
	@Override
	public void clearMsgQueue() {
		Object p = null;
		while ((p = msgQueue.poll()) != null) {
			try {
			} catch (Exception e) {
				System.err.println(e.getMessage());
			}
		}
	}
	@Override
	public void runTask() {
		
		int queueSize = msgQueue.size();
		
		System.out.println("当前队列大小 = "+queueSize + "处理线程 = "+Thread.currentThread().getName());
		if (queueSize == 0) {
			return;
		}
		Object obj = null;
		for (int i = 0; i < queueSize; i++) {
			if ((obj = msgQueue.poll()) != null) {
				System.out.println(Thread.currentThread().getName() +" 获取    " + obj);
			}else{
				break;
			}
		}
	}
}
